package com.citusdata.elven.datastore;

import java.io.UnsupportedEncodingException;
import java.lang.IllegalStateException;


// Provides primary key implementation for the data store. For any entry, the
// partition key and the timestamp are merged together, and they form the
// primary key. Internally, datastore keys are stored as byte arrays, where the
// array holds a fixed 8-byte integer followed by a variable length string.
// However, when comparing datastore keys, the partition key precedes the
// timestamp; most queries access the data in that order, or require that the
// partition key is sorted chronologically. 
// Note that the ordering imposed by this class does not affect how the data is
// stored on disk. For example, the BDB JE implementation is append-only, meaning
// that entries will be stored according to their arrival-order on disk.

public class DatastoreKey {
  private static final String ENCODING  = "UTF-8";
  private static final int    LITTLE_ENDIAN_64_SIZE = 8;

  private final String partitionKey;
  private final long   timestamp;
  private final byte[] bytes;


  public DatastoreKey(String partitionKey, long timestamp) {    
    byte[] keyBytes = null;
    try {
      keyBytes = partitionKey.getBytes(ENCODING);
    } catch (UnsupportedEncodingException ex) {
      // Default encoding should always be supported
      throw new IllegalStateException("Encoding exception: " + ex);
    }

    int offset = LITTLE_ENDIAN_64_SIZE;
    int length = keyBytes.length + LITTLE_ENDIAN_64_SIZE;
    bytes = new byte[length];
    System.arraycopy(keyBytes, 0, bytes, offset, keyBytes.length);
    writeRawLittleEndian64(bytes, timestamp);

    this.partitionKey = partitionKey;
    this.timestamp    = timestamp;
  }


  public DatastoreKey(byte[] bytes) {
    String key = null;
    try {
      int offset = LITTLE_ENDIAN_64_SIZE;
      int length = bytes.length - LITTLE_ENDIAN_64_SIZE;
      key = new String(bytes, offset, length, ENCODING);
    } catch (UnsupportedEncodingException ex) {
      // Default encoding should always be supported
      throw new IllegalStateException("Encoding exception: " + ex);
    }

    this.bytes = bytes;
    this.partitionKey = key;
    this.timestamp    = readRawLittleEndian64(bytes);
  }


  private void writeRawLittleEndian64(byte[] bytes, long value) {
    bytes[0] = (byte) ((int)(value      ) & 0xFF);
    bytes[1] = (byte) ((int)(value >>  8) & 0xFF);
    bytes[2] = (byte) ((int)(value >> 16) & 0xFF);
    bytes[3] = (byte) ((int)(value >> 24) & 0xFF);
    bytes[4] = (byte) ((int)(value >> 32) & 0xFF);
    bytes[5] = (byte) ((int)(value >> 40) & 0xFF);
    bytes[6] = (byte) ((int)(value >> 48) & 0xFF);
    bytes[7] = (byte) ((int)(value >> 56) & 0xFF);
  }


  private long readRawLittleEndian64(byte[] bytes) {
    long value = (((long) bytes[0] & 0xFF)      ) |
                 (((long) bytes[1] & 0xFF) <<  8) |
                 (((long) bytes[2] & 0xFF) << 16) |
                 (((long) bytes[3] & 0xFF) << 24) |
                 (((long) bytes[4] & 0xFF) << 32) |
                 (((long) bytes[5] & 0xFF) << 40) |
                 (((long) bytes[6] & 0xFF) << 48) |
                 (((long) bytes[7] & 0xFF) << 56);
    return value;
  }


  public boolean equals(Object obj) {
    if (!(obj instanceof DatastoreKey)) {
      return false;
    }

    DatastoreKey key = (DatastoreKey) obj;
    return (key.partitionKey.equals(partitionKey)) && (key.timestamp == timestamp);
  }

  public int hashCode() {
    int result = 17;
    result = 37*result + partitionKey.hashCode();
    result = 37*result + (int) (timestamp ^ (timestamp >>> 32));
    return result;
  }

  public String getPartitionKey() {
    return partitionKey;
  }

  public long getTimestamp() {
    return timestamp;
  }

  public byte[] getBytes() {
    return bytes;
  }
}
